package com.sxhuayuan.parking.socket;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.sxhuayuan.parking.exception.NettyException;
import com.sxhuayuan.parking.socket.bo.ParkingLotInterfaceRequest;
import com.sxhuayuan.parking.socket.handler.ParkingLotInterfaceHandler;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelId;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.ChannelGroupFuture;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.GlobalEventExecutor;

/**
 * 通道组池，管理所有websocket连接
 */
public class ChannelGroupPool {

	private static Logger log = LoggerFactory.getLogger(ParkingLotInterfaceHandler.class);

	private Map<String, Channel> channelMap = new HashMap<>();
	// 同步返回结果集
	private Map<String, SyncFuture<String>> syncFutures = new HashMap<>();

	public ChannelGroupPool() {
	}

	public ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

	public boolean addChannel(String username, Channel channel) {
		if (channelMap.containsKey(username)) {
			log.debug("该客户id已登录，强制下线:{}", username);
			Channel channel2 = channelMap.get(username);
			if (channel2.isActive()) {
				ChannelFuture channelFuture = channel2.disconnect();
				channelFuture.addListener(new GenericFutureListener<Future<? super Void>>() {
					@Override
					public void operationComplete(Future<? super Void> future) throws Exception {
						if (future.isDone()) {
							log.debug("客户端已强制下线: {}", username);
						}
					}
				});
			}
		}
		boolean bl = channelGroup.add(channel);
		if (bl) {
			channelMap.put(username, channel);
		}
		return bl;
	}

	public boolean removeChannel(String username) {
		Channel channel = channelMap.get(username);
		channelGroup.remove(channel);
		channelMap.remove(username);
		return true;
	}

	public boolean removeChannel(Channel channel) {
		boolean containsValue = channelMap.containsValue(channel);
		if (containsValue) {
			return channelGroup.remove(channel);
		}
		return false;
	}

	public ChannelFuture sendMsg(String username, String msg) throws NettyException {
		if (!channelMap.containsKey(username)) {
			throw new NettyException();
		}
		log.debug("当前在线客户端:{}", channelMap.keySet());
		Channel channel = channelMap.get(username);
		ChannelFuture future = channel.writeAndFlush(new TextWebSocketFrame(msg));
		return future;
	}

	/**
	 * 发送消息，同步等待返回结果
	 * 
	 * @param username
	 *            车场编号
	 * @param request
	 * @param timeout
	 *            milli
	 * @return
	 * @throws NettyException
	 */
	public String sendMsgSync(String username, ParkingLotInterfaceRequest request, int timeout) throws NettyException {
		if (!channelMap.containsKey(username)) {
			throw new NettyException("客户端不在线: " + username);
		}
		log.debug("当前在线客户端:{}", channelMap.keySet());
		Channel channel = channelMap.get(username);
		String uid = request.getUid();
		if (channel.isActive()) {
			String msg = request.toString();
			log.debug("[{}]TCP send: {}", uid, msg);
			channel.writeAndFlush(new TextWebSocketFrame(msg));
			SyncFuture<String> future = new SyncFuture<>(uid);
			syncFutures.put(uid, future);
			try {
				if (future.await(timeout)) {
					String resp = future.get();
					JSONObject jo = JSON.parseObject(resp);
					return jo.getString("body");
				}
			} catch (InterruptedException | ExecutionException e) {
				log.error("[{}]TCP response timeout", uid, e);
			} finally {
				log.debug("[{}]syncFutures remove", uid);
				syncFutures.remove(uid);
			}
			throw new NettyException(String.format("等待返回超时", uid));
		} else {
			throw new NettyException(String.format("客户端已掉线...", uid));
		}
	}

	/**
	 * 唤醒
	 * 
	 * @param uuid
	 */
	public void notifyResponse(String uuid, String response) {
		SyncFuture<String> future = syncFutures.get(uuid);
		if (future != null) {
			future.setResponse(response);
		} else {
			log.debug("[{}]syncFuture为空", uuid);
		}
	}

	/**
	 * 所有通道
	 * 
	 * @param msg
	 */
	public ChannelGroupFuture sendGroupMsg(String msg) {
		return channelGroup.writeAndFlush(new TextWebSocketFrame(msg));
	}

	public void printDebugInfo() {
		log.debug("allUser={}, syncMap.size={}", channelMap.keySet(), syncFutures.keySet().size());
	}

	public String findUsername(ChannelId channeId) {
		for (Map.Entry<String, Channel> e : channelMap.entrySet()) {
			if (e.getValue().id().equals(channeId)) {
				return e.getKey();
			}
		}
		return null;
	}
}